//package com.ye;
//
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.typeinfo.Types;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.connector.file.src.FileSource;
//import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
//import org.apache.flink.core.fs.Path;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
//import org.apache.flink.util.Collector;
//
//public class WordCountFromText {
//
//    public static void main(String[] args) throws Exception {
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        TextLineInputFormat inputFormat = new TextLineInputFormat();
//        Path path = new Path("D:\\source\\repos\\flink-study\\input\\file.txt");
//        final FileSource<String> fileSource = FileSource.forRecordStreamFormat(
//                inputFormat,
//                path
//                ).build();
////        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "Read Text Files").print();
//        DataStream<Tuple2<String, Integer>> dataStream = env
////                .socketTextStream("localhost", 7777)
//                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "Read Text Files")
////                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
////                    @Override
////                    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
////                        for (String word: s.split(" ")) {
////                            collector.collect(new Tuple2<String, Integer>(word, 1));
////                        }
//////                        String[] split = s.split(" ");
//////                        for (int i = 0; i < split.length; i++) {
//////                           collector.collect(new Tuple2<String, Integer>(split[i], 1));
//////                        }
////                    }
////                })
//                .flatMap((String s, Collector<Tuple2<String, Integer>> collector) -> {
//                    for (String word: s.split(" ")) {
//                        collector.collect(new Tuple2<String, Integer>(word, 1));
//                    }
//                })
//                .returns(Types.TUPLE(Types.STRING, Types.INT))
//                .keyBy(value -> value.f0)
////                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(1);
//
//        dataStream.print();
//
//        env.execute("Window WordCount");
//    }
//}
